package com.atguigu.tingshu.album.service.impl;

import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.cache.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;

	@Autowired
	private AlbumAttributeValueMapper albumAttributeValueMapper;

	@Autowired
	private AlbumStatMapper albumStatMapper;

	@Autowired
	private AlbumAttributeValueService albumAttributeValueService;

	@Autowired
	private KafkaService kafkaService;

	@Autowired
	private RedisTemplate redisTemplate;

	@Autowired
	private RedissonClient redissonClient;

	@Override
	public void updateStat(Long albumId, String statPlay, Integer count) {
		//	调用mapper层
		albumInfoMapper.updateStat(albumId,statPlay,count);
	}

	@Override
	@GuiGuCache(prefix = "albumStat:")
	public AlbumStatVo getAlbumStatVo(Long albumId) {
		//	根据专辑获取统计数据
		return albumInfoMapper.selectAlbumStatVo(albumId);
	}

	@Override
	public List<AlbumAttributeValue> findAlbumAttributeValue(Long albumId) {
		//	select * from album_attribute_value where album_id = ?;
		return albumAttributeValueService.list(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,albumId));
//		return albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,albumId));
	}

	@Override
	public List<AlbumInfo> findUserAllAlbumList(Long userId) {
		//	根据用户Id查询数据。
		//	select id ,album_title from album_info where user_id = ? order by id desc;
		//	SELECT id,album_title FROM album_info WHERE is_deleted=0 AND (user_id = ?) ORDER BY id DESC
		//	SELECT id,album_title FROM album_info WHERE is_deleted=0 AND (user_id = ?) ORDER BY id DESC limit 100
		//	return getAlbumInfoPage();
		//	构建条件查询
		LambdaQueryWrapper<AlbumInfo> wrapper = new LambdaQueryWrapper<>();
		//	user_id = ?
		wrapper.eq(AlbumInfo::getUserId,1);
		//	order by id desc;
		wrapper.orderByDesc(AlbumInfo::getId);
		//	指定sql
		wrapper.last(" limit 100 ");
		//	id ,album_title 设置查询项
		wrapper.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle);

		return albumInfoMapper.selectList(wrapper);
	}

	private List<AlbumInfo> getAlbumInfoPage() {
		Page<AlbumInfo> albumInfoPage = new Page<>(1,100);
		//	构建条件查询
		LambdaQueryWrapper<AlbumInfo> wrapper = new LambdaQueryWrapper<>();
		//	user_id = ?
		wrapper.eq(AlbumInfo::getUserId,1);
		//	order by id desc;
		wrapper.orderByDesc(AlbumInfo::getId);
		//	id ,album_title 设置查询项
		wrapper.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle);
		//	return albumInfoMapper.selectList(wrapper);
		return albumInfoMapper.selectPage(albumInfoPage, wrapper).getRecords();
	}

	@Override
	public void updateAlbumInfo(Long albumId, AlbumInfoVo albumInfoVo) {
		//	先删除缓存的数据： album:info:[1574]
		String key ="album:info:["+albumId+"]";
		//	删除缓存中的数据
		this.redisTemplate.delete(key);

		//	album_info
		AlbumInfo albumInfo = new AlbumInfo();
		//	属性拷贝：
		BeanUtils.copyProperties(albumInfoVo,albumInfo);
		albumInfo.setId(albumId);
		//	更新数据 -- 执行 update 语句.
		albumInfoMapper.updateById(albumInfo);
		//	album_attribute_value  先删除所有数据，再新增数据
		albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,albumId));
		//	获取数据并新增：
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		//	判断集合是否为空。
		if (!CollectionUtils.isEmpty(albumAttributeValueVoList)){
			List<AlbumAttributeValue> attributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
				//	创建专辑属性值对象
				AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
				//	获取到专辑Id 并赋值。先执行的插入，id属性IdType.AUTO
				albumAttributeValue.setAlbumId(albumId);
				BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
				return albumAttributeValue;
			}).collect(Collectors.toList());
			//	保存数据 insert into album_attribute_value values (?,?) (?,?) (?,?) (?,?) (?,?)
			albumAttributeValueService.saveBatch(attributeValueList);
		}

		//	睡眠
		try {
			Thread.sleep(100);
			//	删除数据
			this.redisTemplate.delete(key);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}

		//	是否设置为私密--上架
		if ("1".equals(albumInfoVo.getIsOpen())){
			//	可以发送消息通知检索微服务上架操作. album-->search
			//	根据什么来上架---> 消费者 ---> 调用用法来决定.
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER,String.valueOf(albumId));
		} else {
			//	下架
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER,String.valueOf(albumId));
		}
	}

	@Override
	@GuiGuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX)
	public AlbumInfo getAlbumInfoById(Long albumId) {
		//	return getAlbumInfoByTryLock(albumId);
		return getAlbumInfoDB(albumId);
	}

	private AlbumInfo getAlbumInfoByTryLock(Long albumId) {
		//	定义缓存的key = album:info:albumId
		String albumKey = RedisConstant.ALBUM_INFO_PREFIX + albumId;
		try {
			//	根据key 来获取数据
			AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(albumKey);
			//	String list set ZSet hash{存储对象，修改的时候方便；修改某个属性值的时候，不需要反序列化其他属性。}
			//	判断
			if (null == albumInfo){
				//	使用redis + lua 做分布式锁.
				String albumLocKey = RedisConstant.ALBUM_LOCK_SUFFIX+ albumId;
				//	redissonClient
				RLock lock = this.redissonClient.getLock(albumLocKey);
				//	调用方法
				boolean result = false;
				try {
					result = lock.tryLock(RedisConstant.ALBUM_LOCK_EXPIRE_PX1, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
				} catch (InterruptedException e) {
					throw new RuntimeException(e);
				}
				if (result){
					//	获取到锁
					//	业务逻辑
					try {
						//	缓存中没有数据，查询数据库
						albumInfo = getAlbumInfoDB(albumId);
						//	判断数据库中是否有数据
						if (null == albumInfo){
							//	存储一个空值
							this.redisTemplate.opsForValue().set(albumKey,new AlbumInfo(),RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
							//	删除锁
							return new AlbumInfo();
						}
						//	存储数据
						this.redisTemplate.opsForValue().set(albumKey,albumInfo,RedisConstant.ALBUM_TIMEOUT, TimeUnit.SECONDS);
						//	返回数据
						return albumInfo;
					} catch (Exception e) {
						throw new RuntimeException(e);
					} finally {
						//	删除锁.
						lock.unlock();
					}
				} else {
					try {
						//	没有获取到锁，自旋
						Thread.sleep(300);
						return getAlbumInfoById(albumId);
					} catch (InterruptedException e) {
						throw new RuntimeException(e);
					}
				}
			} else {
				//	缓存中有数据
				return albumInfo;
			}
		} catch (RuntimeException e) {
			//	打印日志.
			log.error("服务器异常",e.getMessage());
			//	发信息.调用短信接口平台，发信息!
			e.getMessage();
		}
		//	如果不走if或else 直接查询数据库.
		return getAlbumInfoDB(albumId);
	}

	private AlbumInfo getAlbumInfoByLock(Long albumId) {
		//	定义缓存的key = album:info:albumId
		String albumKey = RedisConstant.ALBUM_INFO_PREFIX + albumId;
		try {
			//	根据key 来获取数据
			AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(albumKey);
			//	String list set ZSet hash{存储对象，修改的时候方便；修改某个属性值的时候，不需要反序列化其他属性。}
			//	判断
			if (null == albumInfo){
				//	使用redis + lua 做分布式锁.
				String albumLocKey = RedisConstant.ALBUM_LOCK_SUFFIX+ albumId;
				//	redissonClient
				RLock lock = this.redissonClient.getLock(albumLocKey);
				//	调用方法
				lock.lock();
				//	业务逻辑
				try {
					//	获取数据库之前：根据key 来获取数据
					albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(albumKey);
					if (null != albumInfo){
						//	返回数据
						return albumInfo;
					}
					//	缓存中没有数据，查询数据库
					albumInfo = getAlbumInfoDB(albumId);
					//	判断数据库中是否有数据
					if (null == albumInfo){
						//	存储一个空值
						this.redisTemplate.opsForValue().set(albumKey,new AlbumInfo(),RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
						//	删除锁
						return new AlbumInfo();
					}
					//	存储数据
					this.redisTemplate.opsForValue().set(albumKey,albumInfo,RedisConstant.ALBUM_TIMEOUT, TimeUnit.SECONDS);
					//	返回数据
					return albumInfo;
				} catch (Exception e) {
					throw new RuntimeException(e);
				} finally {
					//	删除锁.
					lock.unlock();
				}
			} else {
				//	缓存中有数据
				return albumInfo;
			}
		} catch (RuntimeException e) {
			//	打印日志.
			log.error("服务器异常",e.getMessage());
			//	发信息.调用短信接口平台，发信息!
			e.getMessage();
		}
		//	如果不走if或else 直接查询数据库.
		return getAlbumInfoDB(albumId);
	}

	//	redis + lua 脚本做分布式锁.
	private AlbumInfo getAlbumInfoByRedis(Long albumId) {
		//	定义缓存的key = album:info:albumId
		String albumKey = RedisConstant.ALBUM_INFO_PREFIX + albumId;
		try {
			//	根据key 来获取数据
			AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(albumKey);
			//	String list set ZSet hash{存储对象，修改的时候方便；修改某个属性值的时候，不需要反序列化其他属性。}
		/*
		class Stu{
			private int id ;
			private String name;
		}
		hset key field value; hget key field
		hset stu:1001 id 1001  hset stu:1001 name "atguigu"   hset stu:1001 name "bjatguigu"
		hget stu:1001 id
		hget stu:1001 name

		set key value  get key
		set stu:1001 {"id":"1001","name":"atguigu"}  get  set.setName("bjatguigu"); set stu:1001 {"id":"1001","name":"bjatguigu"}
		stu.getId(); stu.getName();
		 */
			//	判断
			if (null == albumInfo){
				//	使用redis + lua 做分布式锁.
				String albumLocKey = RedisConstant.ALBUM_LOCK_SUFFIX+ albumId;
				//	使用redis命令
				String token = UUID.randomUUID().toString().replaceAll("-","");
				Boolean result = this.redisTemplate.opsForValue().setIfAbsent(albumLocKey, token, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
				//	判断是否获取到锁
				if (result){
					//	缓存中没有数据，查询数据库
					albumInfo = getAlbumInfoDB(albumId);
					if (null == albumInfo){
						//	存储一个空值
						this.redisTemplate.opsForValue().set(albumKey,new AlbumInfo(),RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
						//	删除锁
						delLocKey(albumLocKey, token);
						return new AlbumInfo();
					}
					//	存储数据
					this.redisTemplate.opsForValue().set(albumKey,albumInfo,RedisConstant.ALBUM_TIMEOUT, TimeUnit.SECONDS);
					//	删除锁.
					delLocKey(albumLocKey, token);
					//	返回数据
					return albumInfo;
				} else {
					try {
						//	自旋等待
						Thread.sleep(100);
						return getAlbumInfoById(albumId);
					} catch (InterruptedException e) {
						throw new RuntimeException(e);
					}
				}
			} else {
				//	缓存中有数据
				return albumInfo;
			}
		} catch (RuntimeException e) {
			//	打印日志.
			log.error("服务器异常",e.getMessage());
			//	发信息.调用短信接口平台，发信息!
			e.getMessage();
		}
		//	如果不走if或else 直接查询数据库.
		return getAlbumInfoDB(albumId);
	}

	private void delLocKey(String albumLocKey, String token) {
		//	删除锁
		String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
				"then\n" +
				"    return redis.call(\"del\",KEYS[1])\n" +
				"else\n" +
				"    return 0\n" +
				"end";

		//	创建对象
		DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
		redisScript.setScriptText(scriptText);
		redisScript.setResultType(Long.class);
		this.redisTemplate.execute(redisScript, Arrays.asList(albumLocKey), token);
	}

	//	根据专辑Id 获取专辑数据
	public AlbumInfo getAlbumInfoDB(Long albumId) {
		//	获取专辑对象
		AlbumInfo albumInfo = albumInfoMapper.selectById(albumId);
		//	获取属性信息
		if (null != albumInfo){
			List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
			albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
		}
		//	返回数据
		return albumInfo;
	}

	@Override
	@Transactional(rollbackFor = Exception.class)
	public void removeAlbumInfoById(Long albumId) {
		//	逻辑删除：album_info  album_attribute_value
		albumInfoMapper.deleteById(albumId);
		//	album_stat
		//	表示根据主键Id 删除。非主键 album_id
		//	albumStatMapper.deleteById();
		//	Wrapper 构建条件的 构建where 后面的条件 QueryWrapper UpdateWrapper-设置更新条件，同时还能设置更新值！set where
		//	delete from album_stat where album_id = ?
		//	@TableLogic 这个注解表示逻辑删除。虽然你执行的是del方法()；本质也是update!
		//	第一种：QueryWrapper
		//		QueryWrapper<AlbumStat> wrapper = new QueryWrapper<>();
		//		wrapper.eq("album_id",albumId);
		//	第二种：LambdaQueryWrapper
		LambdaQueryWrapper<AlbumStat> wrapper = new LambdaQueryWrapper<>();
		wrapper.eq(AlbumStat::getAlbumId,albumId);
		//	一个参数
		albumStatMapper.delete(wrapper);
		//	两个参数：第一个参数-要更新的内容，第二个参数更新的条件
		//	update album_stat set is_deleted = 1 where album_id = ?;
		//		AlbumStat albumStat = new AlbumStat();
		//		albumStat.setIsDeleted(1);
		//		albumStatMapper.update(albumStat,wrapper);
		LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
		queryWrapper.eq(AlbumAttributeValue::getAlbumId,albumId);
		albumAttributeValueMapper.delete(queryWrapper);

		kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER,String.valueOf(albumId));
	}

	@Override
	public IPage<AlbumListVo> findUserAlbumPage(Page<AlbumListVo> albumListVoPage, AlbumInfoQuery albumInfoQuery) {
		//	IPage 中 List<AlbumListVo> getRecords() 有很多关于分页的数据条件.
		//	album_info album_stat
		return albumInfoMapper.selectUserAlbumPage(albumListVoPage,albumInfoQuery);
	}

	@Override
	@Transactional(rollbackFor = Exception.class)
	public void saveAlbumInfo(Long userId,AlbumInfoVo albumInfoVo) {
		//	有一些数据是页面没有传递的！user_id tracks_for_free status
		//	创建专辑对象 album_info
		AlbumInfo albumInfo = new AlbumInfo();
		//	赋值：属性拷贝
		BeanUtils.copyProperties(albumInfoVo,albumInfo);
		//	默认值为1，目的是方便查询数据
		albumInfo.setUserId(1l);
		//	审核状态
		albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
		//	判断是否付费：需要付费
		if (!albumInfoVo.getPayType().equals(SystemConstant.ALBUM_PAY_TYPE_FREE)){
			albumInfo.setTracksForFree(5);
		}
		albumInfoMapper.insert(albumInfo);
		//	保存专辑属性 album_attribute_value
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		List<AlbumAttributeValue> albumAttributeValueVoList1  =   albumInfo.getAlbumAttributeValueVoList();
		//	本质：BeanUtils.copyProperties(albumInfoVo,albumInfo); 如果有对象集合; 深拷贝，浅拷贝
		System.out.println("albumInfo:"+albumAttributeValueVoList1);
		//		if (!CollectionUtils.isEmpty(albumAttributeValueVoList)){
		//			//	循环遍历
		//			albumAttributeValueVoList.forEach(albumAttributeValueVo -> {
		//				//	创建专辑属性值对象
		//				AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
		//				//	获取到专辑Id 并赋值。先执行的插入，id属性IdType.AUTO
		//				albumAttributeValue.setAlbumId(albumInfo.getId());
		////				albumAttributeValue.setValueId(albumAttributeValueVo.getValueId());
		////				albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
		//				BeanUtils.copyProperties(albumAttributeValueVo,albumAttributeValue);
		//				//	保存数据 执行很多条insert 语句. 执行5次！
		//				//	可以优化：insert into album_attribute_value values (?,?) (?,?) (?,?) (?,?) (?,?)
		//				//	albumAttributeValueMapper.insert(albumAttributeValue);
		//			});
		//		}
		if (!CollectionUtils.isEmpty(albumAttributeValueVoList)){
			List<AlbumAttributeValue> attributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
				//	创建专辑属性值对象
				AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
				//	获取到专辑Id 并赋值。先执行的插入，id属性IdType.AUTO
				albumAttributeValue.setAlbumId(albumInfo.getId());
				BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
				return albumAttributeValue;
			}).collect(Collectors.toList());
			//	保存数据 insert into album_attribute_value values (?,?) (?,?) (?,?) (?,?) (?,?)
			albumAttributeValueService.saveBatch(attributeValueList);
		}

		//	调用一个方法.album_stat
		this.saveAlbumStat(albumInfo.getId(),SystemConstant.ALBUM_STAT_PLAY);
		this.saveAlbumStat(albumInfo.getId(),SystemConstant.ALBUM_STAT_SUBSCRIBE);
		this.saveAlbumStat(albumInfo.getId(),SystemConstant.ALBUM_STAT_BROWSE);
		this.saveAlbumStat(albumInfo.getId(),SystemConstant.ALBUM_STAT_COMMENT);

		//	是否设置为私密--上架
		if ("1".equals(albumInfoVo.getIsOpen())){
			//	可以发送消息通知检索微服务上架操作. album-->search
			//	根据什么来上架---> 消费者 ---> 调用用法来决定.
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER,String.valueOf(albumInfo.getId()));
		}
	}

	private void saveAlbumStat(Long albumId, String statPlay) {
		//	创建对象
		AlbumStat albumStat = new AlbumStat();
		albumStat.setAlbumId(albumId);
		albumStat.setStatType(statPlay);
		albumStat.setStatNum(0);
		//	保存数据
		this.albumStatMapper.insert(albumStat);
	}
}
